package mapReduce;

import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;

import vo.Comment;


	
public class TopTrendMR {

		public static class Mapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

			protected void map(ImmutableBytesWritable key, Result result,
					Context context) throws IOException, InterruptedException {
				
				String text = Bytes.toString(result.getValue(Bytes.toBytes(Comment.CF_ATTRIBUTES), 
						Bytes.toBytes(Comment.COMMENT_TEXT)));
				
				for (String word : text.split(" ")) {
					ImmutableBytesWritable immutableWord = new ImmutableBytesWritable(Bytes.toBytes(word));
					context.write(immutableWord, immutableWord);
				}
			}

		}
		
		public static class Reducer extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, NullWritable> {

			private static Map<String, Integer> map = new HashMap<String,Integer>();
			
			protected void reduce(ImmutableBytesWritable key, Iterable<ImmutableBytesWritable> it,
					Context context) throws IOException, InterruptedException {
				
				String word = Bytes.toString(key.get());
				int cont = 0;
				for (ImmutableBytesWritable immutableBytesWritable : it) {
					cont++;
				}
				map.put(word, cont);
			}
			
			public static String printTopTrends() {
				List <Map.Entry<String, Integer>> list = new LinkedList<Map.Entry<String, Integer>>(map.entrySet());
				Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {
					public int compare(Entry<String, Integer> e1,
							Entry<String, Integer> e2) {
						return e2.getValue() - e1.getValue();
					}});
				
				return list.size() < 10?list.toString():list.subList(0, 9).toString();
			}

		}}

	